package generic.activemq;

import generic.Subscriber;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMqSubscriber implements MessageListener {

	Session session;
	private Subscriber subscriber;

	
	//private String url = "tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616";
	//private String url = "failover:(tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616,tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61618)?randomize=false";
	//private String url = "failover:(tcp://192.168.11.128:61616,tcp://192.168.11.129:61616)?randomize=false&backup=true&initialReconnectDelay=1&maxReconnectDelay=1&maxReconnectAttempts=1";
	private String url = "failover:(tcp://192.168.11.128:61616,tcp://192.168.11.129:61616)?randomize=false";
	// private String url = "tcp://localhost:61616";
	//private String url = "failover:(tcp://localhost:61616,tcp://localhost:61618)?randomize=false&backup=true&initialReconnectDelay=1&maxReconnectAttempts=1";
	// private String url =
	// "fanout:(static:(tcp://localhost:61616,tcp://localhost:61618))?minAckCount=1&initialReconnectDelay=1&maxReconnectAttempts=1";

	private StringBuilder builder;
	private boolean transacted = true;
	private boolean durable = true;

	public ActiveMqSubscriber(Subscriber subscriber, String url) {
		super();
		this.subscriber = subscriber;
		this.url = url;
		System.out.println("url:" + url);

		builder = new StringBuilder();

		initialize(url);

		System.out.println("ActiveMqListener waiting for messages...");
	}
	
	private void initialize(String brokerUrl) {
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
					brokerUrl);
			Connection connection;
			connection = factory.createConnection();
			connection.setClientID("subscriber1");
			session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
			Topic topic = session.createTopic("topictest.messages");
			MessageConsumer consumer = null;
			if( durable )
				consumer = session.createDurableSubscriber( topic , "subscriber1");
			else
				consumer = session.createConsumer(topic);
			consumer.setMessageListener(this);
			
//			TopicSubscriber subscriber = session.createDurableSubscriber(topic, "myDurable1");
//			subscriber.setMessageListener(this);
			
			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void onMessage(Message jmsMessage) {
		TextMessage message = (TextMessage) jmsMessage;
		try {
			subscriber.receive(message.getText());
            if( transacted ) {
				session.commit();
			}
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
}
